#include "svr_skeleton.h"

struct worker_ctl
{/*{{{*/
    unsigned long thrd_id;
    int worker_idx;
    THRD_MSGQ listener_thmq;
    MSGQ_SEND mqsd; 
    char *respond;
};/*}}}*/
static struct worker_ctl *a_worker_ctl;

struct msgnode
{/*{{{*/
    char from[NAME_SIZE+1];
    unsigned int syn;
    char *msg;
    time_t access_ts;
    struct listp_link link;
};/*}}}*/

int worker_add_msg(int worker_idx, char *from,
        unsigned int syn, char *request)
{/*{{{*/
    struct msgnode *node;
    ALLOC_DOERR(node, struct msgnode,
            errlog("%s: listener alloc msgnode fail, BAD", __func__);
            return(-1);
            );
    strcpy(node->from, from);
    node->syn=syn;
    node->msg=strdup(request);
    if(!node->msg)
    {
        errlog("%s: listener alloc msgnode.msg fail, BAD", __func__);
        free(node);
        return(-1);
    }
    node->access_ts=curr_timestamp(NULL);
    thrd_msgq_send(a_worker_ctl[worker_idx].listener_thmq, &node->link);
    return(0);
}/*}}}*/

static int _proc(struct worker_ctl *worker_ctl, struct msgnode *node)
{/*{{{*/
    static int TIMEOUT=5*1000;
    int worker_idx=worker_ctl->worker_idx;
    char *request=node->msg;
    char *respond=worker_ctl->respond;
    int respond_size;
    int res;

    CLR_BUF(respond);
    char *body=request;
    char *svc;
    if((svc=para_get_item(request, &body)) == NULL ||
            (res=strlen(svc)) <= 0 || res > NAME_SIZE)
    {
        errlog("%s: worker(%d) svc(%s) invalid",
                __func__, worker_idx, (svc)?svc:SPACES);
		para_put_item(respond, SOAR_ERR_SVC_INVALID);
        goto RESPOND;
    }

    struct svc_map *map;
    int svc_idx;
    for(svc_idx=0; svc_idx<SVC_MAP_SIZE; ++svc_idx)
    {
        map=&SVC_MAP[svc_idx];
        if(strcasecmp(map->name, svc) == 0) break;
    }
    if(svc_idx == SVC_MAP_SIZE)
    {
        errlog("%s: worker(%d) proc_(%s) not found",
                __func__, worker_idx, svc);
        para_put_item(respond, SOAR_ERR_SVC_NOT_FOUND);
    }
    else
    {
        map->pfunc(worker_idx, body, respond);
    }

RESPOND:
    respond_size=strlen(respond);
    res=msgq_send(worker_ctl->mqsd, NULL, node->from, node->syn,
            (unsigned char *)respond, respond_size, TIMEOUT);
    if(res)
    {
        errlog("%s: msgq_send fail, res:%d", __func__, res);
        return(-1);
    }
    return(0);
}/*}}}*/

static void _worker(struct worker_ctl *worker_ctl)
{/*{{{*/
#define CHECK_OPERATE_INFO_INTERVAL         1
    int worker_idx=worker_ctl->worker_idx;
    struct listp_link active_list;
    struct msgnode *node;
    unsigned int quit=IS_FALSE;
    int count;
    int res;

    while(!quit)
    {
        res=thrd_msgq_recv(worker_ctl->listener_thmq, &active_list, 500);
        if(res)
        {
            if(res != RET_ERR_TIMEOUT)
            {
                errlog("%s: worker(%d) recv from listener fail, "
                        "res:%d, BAD",
                        __func__, worker_idx, res);
            }
            continue;
        }
        count=0;
        while(!listp_is_empty(&active_list))
        {
            node=listp_entry(active_list.next, struct msgnode, link);
            listp_del(&node->link);
            res=_proc(worker_ctl, node);
            free(node->msg);
            free(node);
            ++count;
        }
        pthread_mutex_lock(&stat_info._worker_.lock);
        struct _worker_info *worker=
            &stat_info._worker_.worker[worker_idx];
        worker->client_num-=count;
        pthread_mutex_unlock(&stat_info._worker_.lock);
    }
}/*}}}*/

int worker_start()
{/*{{{*/
    char buf[200];
    int res;

    // stat_info
    pthread_mutex_init(&stat_info._worker_.lock, NULL);
    ALLOC_MULTI_DOERR(stat_info._worker_.worker, struct _worker_info,
            cnfg_info.start,
            sprintf(buf, "%s: alloc stat_info._worker_.worker fail",
                SERVER_NAME);
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
            );
    int worker_idx;
    for(worker_idx=0; worker_idx<cnfg_info.start; ++worker_idx)
    {
        struct _worker_info *worker=
            &stat_info._worker_.worker[worker_idx];
        worker->client_num=0;
    }

    /*--- thread worker ---*/
    ALLOC_MULTI_DOERR(a_worker_ctl, struct worker_ctl,
            cnfg_info.start,
            sprintf(buf, "%s: alloc a_worker_ctl fail", SERVER_NAME);
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
            );
    for(worker_idx=0; worker_idx<cnfg_info.start; ++worker_idx)
    {
        struct worker_ctl *worker_ctl=&a_worker_ctl[worker_idx];

        worker_ctl->worker_idx=worker_idx;

        // open listener-worker channel
        res=thrd_msgq_create(&worker_ctl->listener_thmq);
        if(res)
        {
            sprintf(buf, "%s: thrd_msgq_create fail, res:%d",
                    __func__, res);
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
        }

        // send msgq
         res=msgq_send_attach(&worker_ctl->mqsd);
         if(res)
         {
             sprintf(buf, "agent: msgq_send_attach fail, res:%d", res);
             fprintf(stderr, "%s\n", buf); errlog(buf);
             return(-1);
         }

        ALLOC_MULTI_DOERR(worker_ctl->respond, char, SOCK_MSG_SIZE,
                sprintf(buf, "%s: alloc worker(%d) respond fail",
                    SERVER_NAME, worker_idx);
                fprintf(stderr, "%s\n", buf); errlog(buf);
                return(-1);
                );

        if(thrd_create((THRD_FUNC_T)_worker, (void *)worker_ctl,
                    THRD_DETACH, &worker_ctl->thrd_id))
        {
            sprintf(buf, "%s: thread worker(%d) create fail",
                    SERVER_NAME, worker_idx);
            fprintf(stderr, "%s\n", buf); errlog(buf);
            return(-1);
        }
    }

    sprintf(buf, "%s: start workers succeed", SERVER_NAME);
    fprintf(stderr, "%s\n", buf); errlog(buf);
    return(0);
}/*}}}*/
